package com.raymond.queue.handler;

import com.alibaba.fastjson.JSONObject;
import com.raymond.queue.FileQueue;
import com.raymond.queue.QueueManager;
import com.raymond.queue.common.msg.request.CreateConsumptionRequest;
import com.raymond.queue.common.msg.response.CreateConsumptionResponse;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * 创建消费组业务处理
 *
 * @author :  raymond
 * @version :  V1.0
 * @date :  2021-07-27 16:37
 */
public class CreateConsumptionRequestHandler extends SimpleChannelInboundHandler<CreateConsumptionRequest> {

    private final static Logger logger = LoggerFactory.getLogger(CreateConsumptionRequestHandler.class);

    @Override
    protected void channelRead0(ChannelHandlerContext ctx, CreateConsumptionRequest msg) throws Exception {
        logger.info("创建消费组请求:{}", JSONObject.toJSONString(msg));
        FileQueue<byte[]> fileQueue = QueueManager.getFileQueue(msg.getTopic());
        if(fileQueue == null) {
            ctx.writeAndFlush(new CreateConsumptionResponse(1, msg.getSequenceId()));
            return;
        }
        try {
            fileQueue.createGroup(msg.getGroupName(), FileQueue.GrowMode.getGrowMode(msg.getGrowMode().getType()),
                    msg.getSrcGroupName(), msg.isContinue());
            ctx.writeAndFlush(new CreateConsumptionResponse(0, msg.getSequenceId()));
        } catch (Exception e) {
            ctx.writeAndFlush(new CreateConsumptionResponse(1, msg.getSequenceId()));
            logger.error("创建消费组失败,请求报文:{}", JSONObject.toJSONString(msg), e);
        }

    }
}
